package spark.yk.java.maven;
import scala.Tuple2;
import scala.collection.mutable.Seq;

import org.apache.spark.SparkConf;  
import org.apache.spark.api.java.JavaPairRDD;  
import org.apache.spark.api.java.JavaRDD;  
import org.apache.spark.api.java.JavaSparkContext;  
import org.apache.spark.api.java.function.FlatMapFunction;  
import org.apache.spark.api.java.function.Function2;  
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
public class SparkWordCount2 {

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		/**
         * 第1步：创建Spark的配置对象SparkConf，设置Spark程序的运行时的配置信息，
         * 例如说通过setMaster来设置程序要链接的Spark集群的Master的URL,如果设置
         * 为local，则代表Spark程序在本地运行，特别适合于机器配置条件非常差（例如 只有1G的内存）的初学者 *
         */
		String uri = "hdfs://192.168.1.185:9000";  
	    String dir = "/user/output1";  
	    String parentDir = "/user";  
	    String remoteFile = uri+dir + "/README.md";
		// TODO Auto-generated method stub
		String logFile = remoteFile; // Should be some file on your system
		SparkConf conf = new SparkConf().setAppName("Simple Application").setMaster("spark://192.168.1.185:7077");
		
		/**
         * 第2步：创建SparkContext对象
         * SparkContext是Spark程序所有功能的唯一入口，无论是采用Scala、Java、Python
         * 、R等都必须有一个SparkContext(不同的语言具体的类名称不同，如果是Java的话则为JavaSparkContext)
         * SparkContext核心作用：初始化Spark应用程序运行所需要的核心组件，包括DAGScheduler、TaskScheduler、
         * SchedulerBackend 同时还会负责Spark程序往Master注册程序等
         * SparkContext是整个Spark应用程序中最为至关重要的一个对象
         */
		@SuppressWarnings("resource")  
        JavaSparkContext sc = new JavaSparkContext(conf);  
		sc.addJar("D:\\Java\\workspace\\SparkJavaDemo\\SparkWordCount2.jar");
		/**
         * 第3步：根据具体的数据来源（HDFS、HBase、Local FS、DB、S3等）通过JavaSparkContext来创建JavaRDD
         * JavaRDD的创建基本有三种方式：根据外部的数据来源（例如HDFS）、根据Scala集合、由其它的RDD操作
         * 数据会被RDD划分成为一系列的Partitions，分配到每个Partition的数据属于一个Task的处理范畴
         * 注意：文件路径不能直接用Windows路径中的反斜扛\，要改成Linux下的斜扛/
         */
        JavaRDD<String> lines = sc.textFile(logFile);
        
        /**
         * 第4步：对初始的JavaRDD进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算
         * 第4.1步：讲每一行的字符串拆分成单个的单词
         */
        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {  
            public Iterator<String> call(String s) {  
              return  Arrays.asList(s.split(" ")).iterator();  
            }  
          });  
            
        /**
         * 第4步：对初始的JavaRDD进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算
         * 第4.2步：在单词拆分的基础上对每个单词实例计数为1，也就是word => (word, 1)
         */
          JavaPairRDD<String, Integer> ones = words.mapToPair(new PairFunction<String, String, Integer>() {  
            public Tuple2<String, Integer> call(String s) {  
              return new Tuple2<String, Integer>(s, 1);  
            }  
          });  
            
          /**
           * 第4步：对初始的RDD进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算
           * 第4.3步：在每个单词实例计数为1基础之上统计每个单词在文件中出现的总次数
           */
          JavaPairRDD<String, Integer> counts = ones.reduceByKey(new Function2<Integer, Integer, Integer>() {  
            public Integer call(Integer i1, Integer i2) {  
              return i1 + i2;  
            }  
          });  
          
        //this method is failure on spark debug
          /*counts.foreach(new VoidFunction<Tuple2<String, Integer>>() {
              public void call(Tuple2<String, Integer> pairs) throws Exception {
                  System.out.println(pairs._1 + " : " + pairs._2);
              }
          });*/
        
          List<Tuple2<String, Integer>> output = counts.collect();  //throw exception 
          for (Tuple2 tuple : output) {  
            System.out.println(tuple._1 + ": " + tuple._2);  
          }  
          System.exit(0);  
	}
}
